package com.atguigu.flink.chapter10;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.*;

/*
用SQL持续查询：
 */
public class TableSqlDemo3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("s1", 1000L, 10),
                new WaterSensor("s2", 2000L, 20),
                new WaterSensor("s1", 3000L, 30),
                new WaterSensor("s1", 4000L, 40),
                new WaterSensor("s2", 5000L, 50)
        );

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(stream);

        //1、查询未注册的表
        //tEnv.executeSql("ddl 增删改");
        //tEnv.sqlQuery("查询语句");

        //Table result = tEnv.sqlQuery("select * from " + table);
        //Table result = tEnv.sqlQuery("select id,sum(vc) as vc_sum from " + table + " group by id");



        //2、查询已注册的表
        tEnv.createTemporaryView("sensor",table);
        tEnv.sqlQuery("select * from sensor").execute().print();

        //result.execute().print();

    }
}


//public class TableSqlDemo3 {
//    public static void main(String[] args) {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//        DataStreamSource<WaterSensor> stream = env.fromElements(
//                new WaterSensor("s1", 1000L, 10),
//                new WaterSensor("s2", 2000L, 20),
//                new WaterSensor("s1", 3000L, 30),
//                new WaterSensor("s1", 4000L, 40),
//                new WaterSensor("s2", 5000L, 50)
//        );
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        Table table = tEnv.fromDataStream(stream);
//
//        //1、查询未注册的表
//        //tEnv.executeSql("ddl 增删改");
//        //tEnv.sqlQuery("查询语句");
//
////        Table result = tEnv.sqlQuery("select * from " + table);
////        Table result = tEnv.sqlQuery("select id,sum(vc) as vc_sum from " + table + " group by id");
////        result.execute().print();
//
//        //2、查询已注册的表
//        tEnv.createTemporaryView("sensor",table);
//        tEnv.sqlQuery("select * from sensor").execute().print();
//    }
//}


//public class TableSqlDemo3 {
//    public static void main(String[] args) {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//        DataStreamSource<WaterSensor> stream = env.fromElements(
//                new WaterSensor("s1", 1000L, 10),
//                new WaterSensor("s2", 2000L, 20),
//                new WaterSensor("s1", 3000L, 30),
//                new WaterSensor("s1", 4000L, 40),
//                new WaterSensor("s2", 5000L, 50)
//        );
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//        Table table = tEnv.fromDataStream(stream);
//
//        //1、查询未注册的表
//        //tEnv.executeSql("ddl 增删改");
//        //tEnv.sqlQuery("查询语句");
//
////        Table result = tEnv.sqlQuery("select * from " + table);
////        Table result = tEnv.sqlQuery("select id,sum(vc) as vc_sum from " + table + " group by id");
////        result.execute().print();
//
//
//        //2、查询已注册的表
//        tEnv.createTemporaryView("sensor",table);
//        tEnv.sqlQuery("select * from sensor").execute().print();
//    }
//}